package com.atguigu.day4

import com.atguigu.source.{SensorReading, SensorSource}
import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.streaming.api.functions.co.CoProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object CoProcessFunctionExceple {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment


    env.setParallelism(1)
    //

    val stream1 = env.addSource(new SensorSource)
    val stream2 = env.fromElements(
      ("sensor_2", 10 * 1000L)
    )

    val result = stream1
      .connect(stream2)
      .keyBy(_.id, _._1) //stream1.id=stream2._1
      .process(new ReadingFilter)

    result.print()


    env.execute()


  }

  class ReadingFilter extends CoProcessFunction[SensorReading, (String, Long), SensorReading] {
    lazy val forwardingEnadled = getRuntimeContext.getState(
      new ValueStateDescriptor[Boolean]("switch", Types.of[Boolean])
    )

    //处理来自传感器的流数据
    override def processElement1(value: SensorReading, ctx: CoProcessFunction[SensorReading, (String, Long), SensorReading]#Context, out: Collector[SensorReading]): Unit = {
      if (forwardingEnadled.value()) {
        out.collect(value)
      }
    }

    //处理来自开关的流数据
    override def processElement2(value: (String, Long), ctx: CoProcessFunction[SensorReading, (String, Long), SensorReading]#Context, out: Collector[SensorReading]): Unit = {
      //打开开关
      forwardingEnadled.update(true)
      //开关元组的第二个值就是放行时间

      val ts = ctx.timerService().currentProcessingTime() + value._2
      ctx.timerService().registerProcessingTimeTimer(ts)

    }

    override def onTimer(timestamp: Long, ctx: CoProcessFunction[SensorReading, (String, Long), SensorReading]#OnTimerContext, out: Collector[SensorReading]): Unit = {

      forwardingEnadled.clear()
    }
  }
}
